package com.zck2

import com.zck2.AggAndFilter.{aggFilterAB, getResFile}
import com.zck2.utils.GetConn
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.scheduler.{SparkListener, SparkListenerStageCompleted, StageInfo}
import java.io.File

object MatchMain {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    val start: Long = System.currentTimeMillis()
    val sc: SparkContext = GetConn.getSc
    sc.addSparkListener(new SparkListener {
      override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {
        val stageInfo: StageInfo = stageCompleted.stageInfo
        val stageId: Int = stageInfo.stageId
        val stageName: String = stageInfo.name
        val stageStartTime: Long = stageInfo.submissionTime.getOrElse(0L)
        val stageEndTime: Long = stageInfo.completionTime.getOrElse(0L)
        val stageDuration: Long = stageEndTime - stageStartTime
      }
    })
    //证认层级
    var hpx: Int = 15    //证认块
    //输入以及输出文件
    val inputPath = args(0)
    val outputPath = modifyPath(inputPath)
    createResDirectory(outputPath)
    println(outputPath)
    //1.获取分区rdd（2个）
    val parRDD = CrossMatch.udfPartitions(sc, inputPath, 15-hpx)     //前面的数为移位块
    //2.证认
    val rddA = CrossMatch.matchAll(parRDD._1, 10)
    val rddB = CrossMatch.matchAll(parRDD._2, 10)
    val rddRes: RDD[(String, Array[String])] = aggFilterAB(rddA, rddB)
    val end1: Long = System.currentTimeMillis()
    val res1: Long = end1 - start
    //3. 获取证认结果
    getResFile(rddRes, outputPath) //tt-hpx、Hpx+Htm证认结果
    val end2: Long = System.currentTimeMillis()
    val res2: Long = end2 - end1
    println("结果文件已保存至："+outputPath)
    sc.stop()
  }

  def modifyPath(inputPath: String): String = {
    val file = new File(inputPath)
    val parentDir = file.getParent
    val lastFolder = file.getName
    val newPath = s"$parentDir/CrossRes/$lastFolder"
    newPath
  }
  def createResDirectory(outputPath: String): Unit = {
    val parentDir = new File(outputPath).getParentFile
    if (!parentDir.exists()) {
      val created = parentDir.mkdirs()
      println("created：是这个！",created)
      if (created) {
        println(s"目录 'CrossRes' 已创建: $parentDir")
      } else {
        println(s"创建父目录失败! $parentDir")
      }
    } else {
      println(s"目录已存在: $parentDir")
    }
  }
}
